package org.example.kafkastudy.config;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.HashMap;
import java.util.Map;

public class ImportantInterceptor<K, V> implements ProducerInterceptor<K, V> {
    private final Map<Long, Long> sendTimes = new HashMap<>();

    @Override
    public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record) {
        System.out.println("ImportantInterceptor...");
        if (record == null) { // 可能前面的拦截器已经拦截过了，会返回 null
            return null;
        }

        // 只有键（Key）为 "important" 的记录才会被发送
        if ("important".equals(record.key())) {
            return record;
        } else {
            return null;
        }
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
    }

    @Override
    public void close() {

    }

    @Override
    public void configure(Map<String, ?> configs) {

    }
}
